package com.sikong.hdfs_tutorial;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.net.URI;
import java.net.URISyntaxException;

/**
 *
 * 这是一个用于演示HDFS的客户端操作Java API操作，包括连接、上传、下载、删除等操作
 * @author 梁超/stan
 * @since 0.1.0
 * @Date 2022/03/04
 */
public class HdfsClient {

    private FileSystem fileSystem;
    private static final Logger LOGGER = LoggerFactory.getLogger(HdfsClient.class);

    /**
     * 用于初始化客户端，包括连接集群等步骤
     */
    public void mkConnection(String hadoopURI, String userName) throws URISyntaxException, IOException, InterruptedException {
        this.fileSystem = FileSystem.get(
                new URI(hadoopURI),
                new Configuration(),
                userName);
    }

    /**
     * 用于判断文件是否存在，方便后续调用，并打印出对应的日志信息
     * @param dst 被判断的文件
     */
    private boolean isExist(String dst) throws IOException {
        if (fileSystem.exists(new Path(dst))) {
            LOGGER.info(String.format("File \"%s\" detected!", dst));
            return true;
        } else {
            LOGGER.info(String.format("File \"%s\" NOT detected!", dst));
            return false;
        }
    }

    /**
     * 用于创建一个文件夹，并打印出对应的日志信息
     * @param fileFolder 用于创建的文件夹
     */
    public void mkDir(String fileFolder) throws IOException {
        if (!isExist(fileFolder)) {
            fileSystem.mkdirs(new Path(fileFolder));
            LOGGER.info(String.format("File \"%s\" successfully created!", fileFolder));
        }
    }

    /**
     * 用于删除一个文件夹，并打印出对应的日志信息
     * @param fileFolder 用于删除的文件夹
     */
    public void delDir(String fileFolder) throws IOException {
        if (isExist(fileFolder)) {
            LOGGER.info(String.format("File \"%s\" Deleting...",fileFolder));
            fileSystem.delete(new Path(fileFolder), true);
            LOGGER.info(String.format("File \"%s\" successfully deleted", fileFolder));
        }
    }

    /**
     * 用于上传一个文件，并打印出对应的日志信息
     * @param localData 本地的文件存储路径
     * @param targetData HDFS上的存储路径
     */
    public void putFile(String localData, String targetData) throws IOException {
        if (!isExist(targetData)) {
            LOGGER.info(String.format("File \"%s\" Uploading...",localData));
            fileSystem.copyFromLocalFile(false, true, new Path(localData), new Path(targetData));
            LOGGER.info(String.format("File \"%s\" successfully uploaded!", localData));
        }
    }

    /**
     * 用于下载一个文件，并打印出对应的日志信息
     * @param targetData HDFS上的存储路径
     * @param localData 本地的文件存储路径
     */
    public void getFile(String targetData, String localData) throws IOException {
        if (!isExist(localData)) {
            LOGGER.info(String.format("File \"%s\" Downloading...",targetData));
            fileSystem.copyToLocalFile(false, new Path(targetData), new Path(localData));
            LOGGER.info(String.format("File \"%s\" successfully downloaded!", targetData));
        }
    }

    /**
     * 用于得到一个文件夹或文件的信息，并打印出对应的日志信息
     * @param dst HDFS上的文件或文件夹存储路径
     */
    public void getInfo(String dst) throws IOException {
        if (isExist(dst)) {
            FileStatus status = fileSystem.getFileStatus(new Path(dst));
            System.out.println(status);
        }
    }

    /**
     * 用于关闭客户端的连接
     */
    public void disConnection() throws IOException {
        fileSystem.close();
        LOGGER.info("HDFS Client disconnected successfully!");
    }

}
